#ifndef __MQ_MESSAGE_HPP__
#define __MQ_MESSAGE_HPP__

#include <iostream>
#include <string>
#include <list>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <sstream>
#include "../mq_comm/mq_msg.pb.h"
#include "../mq_comm/helper.hpp"

namespace mq
{
/*消息持久化操作集合*/
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
    /*消息持久化文件格式=basedir指定目录+qname队列名称+指定后缀*/
    using MessagePtr = std::shared_ptr<Message>;
    class MessageMapper
    {
    public:
        MessageMapper(std::string &basedir, const std::string &qname)
            : _qname(qname)
        {
            if (basedir.back() != '/')
                basedir += '/';
            _datafile = basedir + _qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + _qname + TMPFILE_SUBFIX;
            if (FileHelper(basedir).exists() == false)
                assert(FileHelper::createDirectory(basedir));
            createMsgFile();
        }

        /*创建消息持久化文件*/
        bool createMsgFile()
        {
            if (FileHelper(_datafile).exists() == true)
                return true;
            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                ELOG("创建队列消息文件 %s 失败", _datafile.c_str());
                return false;
            }
            return true;
        }

        /*删除消息持久化文件*/
        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }

        /*向消息持久化文件中插入数据*/
        bool insert(MessagePtr &msg)
        {
            return insert(_datafile, msg);
        }

        /*从消息持久化文件中删除指定内容*/
        bool remove(MessagePtr &msg)
        {
            // 1.修改有效位为0
            msg->mutable_payload()->set_valid("0");
            // 2.序列化msg的消息属性
            std::string body = msg->payload().SerializeAsString();
            if (body.size() != msg->length())
            {
                ELOG("不能修改文件中的数据信息,因为新生成的数据与原数据长度不一致");
                return false;
            }
            // 3.向指定偏移量处写入数据
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if (ret == false)
            {
                ELOG("向消息持久化文件中写入数据失败");
                return false;
            }
            return true;
        }

        /*垃圾回收或恢复历史数据*/
        std::list<MessagePtr> gc()
        {
            // 1.加载所有有效数据
            std::list<MessagePtr> result;
            bool ret = load(result);
            if (ret == false)
            {
                ELOG("加载有效数据失败");
                return result;
            }
            // 2.将有效数据都写入到临时文件中
            FileHelper::createFile(_tmpfile);
            for (auto &iter : result)
            {
                ret = insert(_tmpfile, iter);
                if (ret == false)
                {
                    ELOG("向临时文件写入失败");
                    return result;
                }
            }
            // 3.删除源文件
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                ELOG("删除源文件失败");
                return result;
            }
            // 4.修改临时文件名为源文件名
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                ELOG("修改临时文件名称失败");
                return result;
            }
            return result;
        }

    private:
        bool insert(const std::string &filename, MessagePtr &msg)
        {
            // 1.序列化消息属性
            std::string body = msg->payload().SerializeAsString();
            // 2.获取文件长度
            FileHelper helper(filename);
            size_t fsize = helper.size();
            size_t msg_size = body.size();
            // 3.先向文件中写入4字节长度
            bool ret = helper.write((char *)&msg_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                ELOG("向队列消息持久化文件 %s 中写入长度数据失败", filename.c_str());
                return false;
            }
            // 4.写入正文内容
            ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
            if (ret == false)
            {
                ELOG("向队列消息持久化文件 %s 中写入内容失败", filename.c_str());
                return false;
            }
            // 5.更新msg字段
            msg->set_offset(fsize + sizeof(size_t));
            msg->set_length(body.size());
            return true;
        }

        /*被gc()调用的接口，主要是用于获取当前文件中有效的数据*/
        bool load(std::list<MessagePtr> &result)
        {
            // 1.获取文件中所有有效数据
            FileHelper helper(_datafile);
            size_t fsize = helper.size();
            size_t index = 0;
            while (index < fsize)
            {
                // 2.先获取四字节长度
                size_t msg_size = 0;
                bool ret = helper.read((char *)&msg_size, index, sizeof(size_t));
                if (ret == false)
                {
                    ELOG("读取消息长度失败");
                    return false;
                }
                // 3.获取具体内容
                index += sizeof(size_t);
                std::string body(msg_size, '\0');
                ret = helper.read(&body[0], index, msg_size);
                if (ret == false)
                {
                    ELOG("读取消息内容失败");
                    return false;
                }
                // 4.序列化填充
                index += msg_size;
                MessagePtr msg = std::make_shared<Message>();
                msg->mutable_payload()->ParseFromString(body);
                if (msg->payload().valid() == "0")
                {
                    DLOG("加载到无效消息: %s", msg->payload().body().c_str());
                    continue;
                }
                // 5.插入list
                result.push_back(msg);
            }
            return true;
        }

    private:
        std::string _qname;    // 队列名称
        std::string _datafile; // 消息持久化文件名
        std::string _tmpfile;  // 中间临时文件名
    };

    /*
        消息队列类
        这里阐述的就是实际消息队列,以链表list管理消息,形成认知上的消息队列
    */
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string &basedir, const std::string &qname)
            : _qname(qname), _valid_count(0), _total_count(0), _mapper(basedir, qname)
        {
        }

        /*获取历史数据*/
        bool recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc();
            for (auto &iter : _msgs)
            {
                _durable_msgs.insert(std::make_pair(iter->payload().properties().id(), iter));
            }
            _valid_count = _total_count = _msgs.size();
            return true;
        }

        /*向队列中插入消息*/
        bool insert(const BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // 1.新建消息
            MessagePtr msg = std::make_shared<Message>();
            msg->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }

            // 2.判断是否需要持久化
            std::unique_lock<std::mutex> locK(_mutex);
            if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                msg->mutable_payload()->set_valid("1");
                // 3.进行持久化存储
                bool ret = _mapper.insert(msg);
                if (ret == false)
                {
                    ELOG("持久化存储消息: %s 失败了", body.c_str());
                    return false;
                }
                _valid_count += 1;
                _total_count += 1;
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            // 4.内存管理
            _msgs.push_back(msg);
            return true;
        }

        /*返回队列首部消息,并推出消息*/
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_msgs.size() == 0)
                return MessagePtr();
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }

        /*
            根据消息ID从待确认消息哈希中移除消息  效果为确认消息
            每次进行remove确认消息的时候,删除消息持久化文件消息时,都会根据要求进行一次垃圾回收,整理一下消息持久化文件内容
        */
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1.从待确认消息哈希中查找
            auto iter = _waitack_msgs.find(msg_id);
            if (iter == _waitack_msgs.end())
            {
                DLOG("没有找到要确认的消息: %s", msg_id.c_str());
                return false;
            }
            // 2.根据消息本身durable判断是否从消息持久化文件中删除
            if (iter->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 3.从消息持久性文件中移除消息
                _mapper.remove(iter->second);
                _durable_msgs.erase(msg_id);
                _valid_count -= 1; // 持久化文件中的有效消息数量-1
                gc();              // 根据要求进行垃圾回收
            }
            // 4.从待确认消息哈希中删除
            _waitack_msgs.erase(msg_id);
            return true;
        }

        /*获取待推送消息数量*/
        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }

        /*获取消息持久化文件中消息总数量*/
        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }

        /*获取持久哈希中消息数量*/
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        /*获取待确认哈希消息数量*/
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }

        /*消息最终清理*/
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1.文件清理
            _mapper.removeMsgFile();
            // 2.内存清理
            _valid_count = _total_count = 0;
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
        }

    private:
        /*判断当前是否满足回收策略*/
        bool GCCheck()
        {
            if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
                return true;
            return false;
        }
        /*当持久化文件中总消息数量超过2000并且有效比例低于50%,启动垃圾回收策略*/
        void gc()
        {
            // 1.判断是否符合gc条件
            if (GCCheck() == false)
                return;
            // 2.开始回收垃圾
            std::list<MessagePtr> msgs = _mapper.gc(); // 回收成功
            // 3.修改持久化消息哈希内容
            for (auto &msg : msgs)
            {
                auto iter = _durable_msgs.find(msg->payload().properties().id());
                if (iter == _durable_msgs.end())
                {
                    DLOG("垃圾回收后,有一条持久化消息,在内存中没有进行管理!");
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                    continue;
                }
                // 4.更新当前实际位置
                iter->second->set_offset(msg->offset());
                iter->second->set_length(msg->length());
            }
            // 4.更新当前持久化的有效消息数量和总的持久化消息数量
            _valid_count = _total_count = msgs.size();
        }

    private:
        std::mutex _mutex;                                         // 互斥锁
        std::string _qname;                                        // 队列名称
        size_t _valid_count;                                       // 消息持久化文件中有效消息数量
        size_t _total_count;                                       // 消息持久化文件中总的消息数量
        MessageMapper _mapper;                                     // 操作消息持久化对象
        std::list<MessagePtr> _msgs;                               // 待推送消息
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息哈希
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息哈希
    };

    /*消息管理类*/
    class MessageManager
    {
    public:
        using ptr=std::shared_ptr<MessageManager>;
        MessageManager(const std::string &basedir) : _basedir(basedir) {}

        /*根据队列名称初始化加载指定队列的持久化消息文件信息*/
        void initQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp = nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto iter = _queue_msgs.find(qname);
                if (iter != _queue_msgs.end())
                    return; // 已经存在了,就不用初始化加载了
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            qmp->recovery();
        }

        /*根据队列名称销毁指定消息队列*/
        void destroyQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp = nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto iter = _queue_msgs.find(qname);
                if (iter == _queue_msgs.end())
                    return; // 本身不存在消息队列,就不用销毁了
                qmp = iter->second;
                _queue_msgs.erase(iter);
            }
            qmp->clear();
        }

        /*向指定队列中插入消息*/
        bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            QueueMessage::ptr qmp = nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto iter = _queue_msgs.find(qname);
                if (iter == _queue_msgs.end())
                {
                    ELOG("向队列%s中新增消息失败:没有找到消息管理句柄!", qname.c_str());
                    return false;
                }
                qmp = iter->second;
            }
            return qmp->insert(bp, body, queue_is_durable);
        }

        /*获取指定队列中的消息,并推出队列消息*/
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr qmp = nullptr;
            {
                std::unique_lock<std::mutex> locK(_mutex);
                auto iter = _queue_msgs.find(qname);
                if (iter == _queue_msgs.end())
                {
                    ELOG("向队列%s中获取消息失败:没有找到消息管理句柄!", qname.c_str());
                    return MessagePtr();
                }
                qmp = iter->second;
            }
            return qmp->front();
        }

        /*根据队列名称向指定队列中确认指定的消息*/
        bool ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmp = nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto iter = _queue_msgs.find(qname);
                if (iter == _queue_msgs.end())
                {
                    ELOG("向队列%s中确认消息失败:没有找到消息管理句柄!", qname.c_str());
                    return false;
                }
                qmp = iter->second;
            }
            return qmp->remove(msg_id);
        }

        /*根据队列名称获取指定队列中待推送消息数量*/
        size_t getable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp = nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto iter = _queue_msgs.find(qname);
                if (iter == _queue_msgs.end())
                {
                    ELOG("向队列%s中获取待推送消息数量失败:没有找到消息管理句柄!", qname.c_str());
                    return false;
                }
                qmp = iter->second;
            }
            return qmp->getable_count();
        }

        /*根据队列名称获取指定队列中消息持久化文件中消息总数量*/
        size_t total_count(const std::string &qname)
        {
            QueueMessage::ptr qmp = nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto iter = _queue_msgs.find(qname);
                if (iter == _queue_msgs.end())
                {
                    ELOG("向队列%s中获取消息持久化文件中消息总数量失败:没有找到消息管理句柄!", qname.c_str());
                    return false;
                }
                qmp = iter->second;
            }
            return qmp->total_count();
        }

        /*根据队列名称获取指定队列中持久化消息数量*/
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp = nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto iter = _queue_msgs.find(qname);
                if (iter == _queue_msgs.end())
                {
                    ELOG("向队列%s中获取持久化消息数量失败:没有找到消息管理句柄!", qname.c_str());
                    return false;
                }
                qmp = iter->second;
            }
            return qmp->durable_count();
        }

        /*根据队列名称获取指定队列中待确认消息数量*/
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmp = nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto iter = _queue_msgs.find(qname);
                if (iter == _queue_msgs.end())
                {
                    ELOG("向队列%s中获取待确认消息数量失败:没有找到消息管理句柄!", qname.c_str());
                    return false;
                }
                qmp = iter->second;
            }
            return qmp->waitack_count();
        }

        /*清理所有消息队列*/
        void clear()
        {
            std::unique_lock<std::mutex> locK(_mutex);
            for(auto& iter:_queue_msgs)
            {
                iter.second->clear();
            }
        }

    private:
        std::mutex _mutex;                                              // 互斥锁
        std::string _basedir;                                           // 消息持久化文件存放目录
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs; // 哈希管理消息
    };
}

#endif